package com.free.bsf.metric;

import com.free.bsf.core.util.PropertyUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.logging.log4j.util.PropertiesUtil;

import java.util.Properties;

public class Kafka {
    private static KafkaProducer<Integer, String> producer;
    public static KafkaProducer<Integer, String> createProducer(){
        if(producer==null) {
            Properties properties = new Properties();
            // 若要配多个服务器，用逗号隔开
            // 注：服务器要开放端口，若云服务器还要在server.properties配置内网IP和外网IP
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, PropertyUtils.getPropertyCache("kafka.servers",""));
            properties.put(ProducerConfig.CLIENT_ID_CONFIG, "metric-producer");
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            // 构造Client无非是：new 或 工厂模式
            producer = new KafkaProducer<>(properties);
        }
        return producer;
    }

    public static void sendMessage(String topic,String msg){
        createProducer().send(new ProducerRecord<>(topic, msg));
    }
}
